//package com.lagou.bak;
//
//import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
//
//public class StateTest2 {
//    public static void main(String[] args) {
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        KeyedStream<Action, Long> actionsByUser = actions
// .keyBy((KeySelector<Action, Long>) action -> action.userId);
//       MapState
//MapStateDescriptor<Void, Pattern> bcStateDescriptor =
// new MapStateDescriptor<>(  "patterns", Types.VOID, Types.POJO(Pattern.class));
//       Pattern MapState Void Pattern Pattern MapState null
//BroadcastStream<Pattern> bcedPatterns = patterns.broadcast(bcStateDescriptor);
//       MapStateDescriptor``broadcast()``patterns``BroadcastStream bcedPatterns
//DataStream<Tuple2<Long, Pattern>> matches = actionsByUser
// .connect(bcedPatterns)
// .process(new PatternEvaluator());
//       actionsByUser``bcedPatterns``connect()``PatternEvaluator`在连接的流上。是一个实现接口的自定义函数。它应用我们之前讨论过的模式匹配逻辑，并发出包含用户ID和匹配模式的记录。该 接口提供了三种处理记录和发出结果的方法。`PatternEvaluator``KeyedBroadcastProcessFunction``Tuple2<Long, Pattern>
//       public static class PatternEvaluator`
// extends KeyedBroadcastProcessFunction<Long, Action, Pattern, Tuple2<Long, Pattern>> {
//
// // handle for keyed state (per user)`
// ValueState<String> prevActionState;`
//
// @Override
// public void open(Configuration conf) {
//  // initialize keyed state`
//  prevActionState = getRuntimeContext().getState(
//                         new ValueStateDescriptor<>("lastAction", Types.STRING));
// }
// /**`
//            ` * Called for each user action.`
//            ` * Evaluates the current pattern against the previous and`
//            ` * current action of the user.`
//            ` */
// @Override`
// public void processElement(
//   Action action,
//  ReadOnlyContext ctx,
//   Collector<Tuple2<Long, Pattern>> out) throws Exception {
//  // get current pattern from broadcast state`
// Pattern pattern = ctx
//   .getBroadcastState(
//                           new MapStateDescriptor<>("patterns", Types.VOID, Types.POJO(Pattern.class)))
//   // access MapState with null as VOID default value`
//   .get(null);`
//  // get previous action of current user from keyed state`
//  String prevAction = prevActionState.value();
//  if (pattern != null && prevAction != null) {
//   // user had an action before, check if pattern matches`
//   if (pattern.firstAction.equals(prevAction) &&
//     pattern.secondAction.equals(action.action)) {
//    // MATCH`
//    out.collect(new Tuple2<>(ctx.getCurrentKey(), pattern));
//   }
//  }
//  // update keyed state and remember action for next pattern evaluation`
//  prevActionState.update(action.action);
// }
// /**`
//            ` * Called for each new pattern.`
//            ` * Overwrites the current pattern with the new pattern.`
//            ` */
// @Override
// public void processBroadcastElement(
//   Pattern pattern,
//   Context ctx,
//   Collector<Tuple2<Long, Pattern>> out) throws Exception {
//  // store the new pattern by updating the broadcast state`
//  BroadcastState<Void, Pattern> bcState =`
//   ctx.getBroadcastState(new MapStateDescriptor<>("patterns", Types.VOID, Types.POJO(Pattern.class)));
//  // storing in MapState with null as VOID default value`
//  bcState.put(null, pattern);
// }
//}
//        KeyedBroadcastProcessFunction
//    }
//}
